import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.configuration.MemorySize;
import org.apache.flink.configuration.RestOptions;
import org.apache.flink.configuration.TaskManagerOptions;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.streaming.util.serialization.SimpleStringSchema;

import java.math.BigDecimal;
import java.util.Properties;

/**
 * @author LiuMingxin
 * @date 2024/11/15&17:17
 */
public class FlinkRunner {
    public static BigDecimal ONE_BIGDECIMAL = new BigDecimal("1");

    public static void main(String[] args) throws Exception {
        org.apache.flink.configuration.Configuration configuration = new org.apache.flink.configuration.Configuration();
        configuration.setInteger(RestOptions.PORT, 9999);
        configuration.set(TaskManagerOptions.NETWORK_MEMORY_MIN, MemorySize.parse("3g"));
        configuration.set(TaskManagerOptions.NETWORK_MEMORY_MAX, MemorySize.parse("3g"));
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(configuration);

        Properties properties = new Properties();
        properties.setProperty("bootstrap.servers", "127.0.0.1:9092");
        properties.setProperty("group.id", "lm-group");
        properties.setProperty("auto.offset.reset", "earliest");

        // 创建Kafka消费者
        FlinkKafkaConsumer<String> consumer = new FlinkKafkaConsumer<>(
                "LMTopic", // Kafka主题
                new SimpleStringSchema(), // 数据序列化方式
                properties // Kafka消费者配置
        );

        DataStream<String> stream = env.addSource(consumer);

        // 对读取的数据进行简单的转换操作
        DataStream<String> mappedStream = stream.map(new MapFunction<String, String>() {
            @Override
            public String map(String value) throws Exception {
                JSONObject jsonObject = JSONObject.parseObject(value);
                jsonObject.put("p1", jsonObject.getBigDecimal("p1").add(ONE_BIGDECIMAL));
                jsonObject.put("p2", jsonObject.getBigDecimal("p2").add(ONE_BIGDECIMAL));
                jsonObject.put("p3", jsonObject.getBigDecimal("p3").add(ONE_BIGDECIMAL));
                jsonObject.put("p4", jsonObject.getBigDecimal("p4").add(ONE_BIGDECIMAL));
                jsonObject.put("p5", jsonObject.getBigDecimal("p5").add(ONE_BIGDECIMAL));
                jsonObject.put("p6", jsonObject.getBigDecimal("p6").add(ONE_BIGDECIMAL));
                return "Flink Get Kafka Message: " + jsonObject.toJSONString();
            }
        });

        // 打印结果到控制台
        mappedStream.print();

        env.execute("Flink Kafka Example");
    }

}
